package cn.doitedu

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

import scala.collection.mutable.ListBuffer

case class Event(guid: Long, event_id: String, action_time: Long)

case class AttributeResult(model_id: String,
                           target_event: String,
                           strategy_name: String,
                           guid: Long,
                           target_event_time: Long,
                           attribute_event: String,
                           attribute_event_time: Long,
                           attribute_ratio: Float
                          )

object Job04_归因分析_主题聚合表计算 {

  def main(args: Array[String]): Unit = {

    // 业务目标事件
    var targetEventId = args(0)

    // 传入的待归因事件
    var attributeEvents = args(1).split(",")

    // [a,b,c]=> ['a','b','c']=> 'a','b','c'
    val str = attributeEvents.map(s => "'" + s + "'").mkString(",")
    val inParam = str + ",'" + targetEventId + "'"


    val spark = SparkSession
      .builder()
      .appName("归因分析计算")
      .master("local")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._


    // 读数据，过滤出归因模型需求中所限定的用户行为
    val df = spark.sql(
      s"""
         |
         |select
         |    guid,event_id,action_time
         |from tmp.user_action_detail
         |where dt='2024-03-31'
         |and event_id in ($inParam)
         |
         |""".stripMargin)


    val ds: Dataset[Event] = df.as
    val rdd: RDD[Event] = ds.rdd

    val resultRdd = rdd.groupBy(event => event.guid)
      .map(tp => tp._2)
      .flatMap(iter => {
        // 比如 : 目标事件: e_x
        // 待归因事件：  e1,e2,e5
        // [ Event(1,e1,10), Event(1,e1,11), Event(1,e2,13), Event(1,e5,16), Event(1,e4,17), Event(1,e2,19) , Event(1,e4,20)]
        // e1,e1,e2,e5,e_x  || e2,e_x, e2
        val events = iter.toList

        val results = ListBuffer.empty[AttributeResult]

        // 1. 先去掉最后一个目标事件之后的 归因事件
        val lastTargetEventIndex = events.lastIndexWhere(e => e.event_id.equals(targetEventId))
        if (lastTargetEventIndex >= 0) {
          // 2. 截断行为事件list
          // e1,e1,e2,e5,e_x,   e2,e_x
          val trimEvents = events.slice(0, lastTargetEventIndex + 1)

          val seg = ListBuffer.empty[Event]

          for (i <- 0 until trimEvents.size) {
            val curEvent = trimEvents(i)
            // 遇到一个目标事件，就分一段
            if (curEvent.event_id.equals(targetEventId)) {
              // 对本段进行计算
              if (seg.nonEmpty) {
                seg += curEvent

                val sortedSeg = seg.sortBy(event => event.action_time).toList
                results += firstAttribute(sortedSeg, targetEventId)
                results += lastAttribute(sortedSeg, targetEventId)
                results ++= linearAttribute(sortedSeg, targetEventId)
                results ++= timeDecayAttribute(sortedSeg, targetEventId)

                seg.clear()
              }

            } else {
              seg += curEvent
            }
          }
        }

        results
      })

    // 把结果转成表
    val resultDs = spark.createDataset(resultRdd)
    resultDs.createOrReplaceTempView("result_view")
    spark.sql(
      """
        |
        |insert into dws.attribute_ratio_agg partition(dt='2024-03-31')
        |select
        |   *
        |from result_view
        |
        |""".stripMargin)


  }

  // 首次触点归因策略计算
  def firstAttribute(seg: List[Event], targetEventId: String): AttributeResult = {
    // seg: [ (Event(3,e1,11), Event(3,e2,12), Event(3,e1,13) ,Event(3,e_x,14) ]
    val event = seg.head
    AttributeResult("model-003", targetEventId, "首次触点归因策略", event.guid, seg.last.action_time, event.event_id, event.action_time, 1.0f)
  }

  // 末次触点归因策略计算
  def lastAttribute(seg: List[Event], targetEventId: String): AttributeResult = {

    val targetEventTime = seg.last.action_time
    val last = seg.slice(0, seg.size - 1).last

    AttributeResult("model-003", targetEventId, "末次触点归因策略", last.guid, targetEventTime, last.event_id, last.action_time, 1.0f)
  }

  // 线性归因策略计算
  def linearAttribute(seg: List[Event], targetEventId: String): List[AttributeResult] = {
    // seg: [ (Event(3,e1,11), Event(3,e2,12), Event(3,e1,13) ,Event(3,e_x,14) ]

    val targetEventTime = seg.last.action_time

    // 保留2位小数
    val ratio = ((100.0 / (seg.size - 1)).round) / 100.0f

    for (elem <- seg.slice(0, seg.size - 1)) yield {
      AttributeResult("model-003", targetEventId, "线性归因策略", elem.guid, targetEventTime, elem.event_id, elem.action_time, ratio)
    }

  }


  // 时间衰减策略计算
  def timeDecayAttribute(seg: List[Event], targetEventId: String): List[AttributeResult] = {
    // seg: [ (Event(3,e1,11), Event(3,e2,12), Event(3,e1,13)  ,Event(3,e_x,14) ]

    // 取出目标事件的行为时间
    val targetEventTime = seg.last.action_time
    // 取出所有待归因事件
    val events = seg.slice(0, seg.size - 1)

    // 求待归因之间的索引之和
    val idxSum = events.indices.map(idx => idx + 1).sum


    val tmp = for (idx <- 0 until events.size) yield {
      val elem = events(idx)
      val ratio = ( (idx+1) * 100.0f / idxSum).round / 100f

      AttributeResult("model-003", targetEventId, "时间衰减归因策略", elem.guid, targetEventTime, elem.event_id, elem.action_time, ratio)
    }

    tmp.toList

  }


}
